package com.apex.spark.source;

import com.apex.spark.SparkEnvironment;
import com.apex.spark.structuredstream.SparkStructuredStreamingSource;
import com.typesafe.config.Config;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.streaming.StreamingContext;

public class SocketSource implements SparkStructuredStreamingSource {

    private Config config;

    private StreamingContext streamingContext;

    @Override
    public void prepare(SparkEnvironment environment) {
    }

    @Override
    public Config getConfig() {
        return config;
    }

    @Override
    public void setConfig(Config config) {
       this.config = config;
    }

    @Override
    public Dataset<Row> getData(SparkEnvironment env) {
        Dataset<Row> lines = env.getSparkSession()
                .readStream()
                .format("socket")
                .option("host", "localhost")
                .option("port", 9999)
                .load();
        return lines;
    }
}
